package cn.ipanel.bigdata.job.realtime

import cn.ipanel.bigdata.boot.Job
import cn.ipanel.bigdata.boot.date.{Date, Hour, Minute}
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.dw.realtime.{T_DEVICE_STATUS => T, T_ONLINE_DEVICE_CT => CT}
import cn.ipanel.bigdata.dw.dim.phoenix.{T_AREA => A}
import cn.ipanel.bigdata.dw.dim.phoenix.{T_DEVICE => D}
import cn.ipanel.bigdata.dw.ods.{T_REPORT => R}
import cn.ipanel.bigdata.utils.Dictionary.BusState
import org.apache.spark.sql.functions.{col, count, lit, sum, when}

/**
 * @author lzz
 * @environment IntelliJ IDEA 2020.3.1
 * @projectName bigdata_shanxi
 * @date 2023/03/28 16:52
 * @description: 实时在线设备统计，小时：只要整点，在线的就算
 */
class Realtime60 extends Job {

  var hour :Hour = _

  override def onPrepare(): Unit = {
    hour = Date.asHour.prev()
    Logger.I("hour: " + hour.toSimpleHour)
    CT.deleteByPeriodTime(hour.toSimpleHour.toInt)
  }

  override def onStartup(): Unit = {


    val res = T.load
      // 姑且认为是在小时结束后，来查看本小时内的状态
      // update_time  status
      // 12: 30       1
      // 13: 00       0
      // 14: 10       1
      // 当前小时离线：status == 0 and update_time <= hour.st
      .filter(col(T.F_STATUS) === lit(T.F_STATUS_ONLINE) or col(T.F_UPDATE_TIME) > lit(hour.st))
      .select(T.F_DEVICE_ID)
      .withColumnRenamed(T.F_DEVICE_ID, D.F_DEVICE_ID)
      .withColumn("online", lit(1))
      .join(D.load.filter(col(D.F_STATUS) =!= lit(BusState.DEL)), Seq(D.F_DEVICE_ID), "right")
      .select(D.F_DEVICE_ID, D.F_REGION, "online")
      .groupBy(D.F_REGION)
      .agg(
        sum(when(col("online") === lit(1), col("online")).otherwise(lit(0))) as CT.F_ONLINE_NUM,
        sum(when(col("online") === lit(1), lit(0)).otherwise(lit(1))) as CT.F_OFFLINE_NUM
      )
      .withColumn(CT.F_PERIOD_TIME, lit(hour.toSimpleHour))

    res.persist()
    res.show(false)
    CT.save(res)

  }

}